package priv.tuyou.myself;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class StreamWindowTest {

	public static void main(String[] args) throws Exception {

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		env.socketTextStream("192.168.142.129", 8888).flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
			@Override
			public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
				out.collect(new Tuple2<>(value, 1));
			}
		}).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Tuple2<String, Integer>>() {
			@Override
			public long extractAscendingTimestamp(Tuple2<String, Integer> element) {
				return (int) element.getField(1);
			}
		})
				// .assignTimestampsAndWatermarks(
				// new
				// BoundedOutOfOrdernessTimestampExtractor<Tuple2<String,Integer>>(Time.milliseconds(1))
				// {
				// @Override
				// public long extractTimestamp(Tuple2<String, Integer> element)
				// {
				// return (int)element.getField(1);
				// }
				// })
				.keyBy(0)
				// .window(TumblingEventTimeWindows.of(Time.seconds(1)))
				.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
				// .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
				// .window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))
				// .window(GlobalWindows.create())
				// .reduce(new ReduceFunction<Tuple2<String,Integer>>() {
				//
				// @Override
				// public Tuple2<String, Integer> reduce(Tuple2<String, Integer>
				// value1, Tuple2<String, Integer> value2)
				// throws Exception {
				//
				// Tuple2<String, Integer> tuple = new Tuple2<String,
				// Integer>(value1.getField(0), (int)value1.getField(1) +
				// (int)value2.getField(1));
				// return tuple;
				// }
				// })
				// .apply(new WindowFunction<Tuple2<String,Integer>, String,
				// Tuple, TimeWindow>() {
				//
				// //一次性iterator，所以有性能和资源开销
				// @Override
				// public void apply(Tuple key, TimeWindow window,
				// Iterable<Tuple2<String, Integer>> input,
				// Collector<String> out) throws Exception {
				//
				// System.out.println("--------------------------");
				// input.forEach(item -> {
				// System.out.println(item.getField(0) + "--" +
				// item.getField(1));
				// });
				// }
				// })
				// .process(new ProcessWindowFunction<Tuple2<String,Integer>,
				// String, Tuple, TimeWindow>() {
				// @Override
				// public void process(Tuple key,
				// ProcessWindowFunction<Tuple2<String, Integer>, String, Tuple,
				// TimeWindow>.Context context,
				// Iterable<Tuple2<String, Integer>> elements, Collector<String>
				// out) throws Exception {
				//
				// System.out.println("--------------------------");
				// elements.forEach(item -> {
				// System.out.println(item.getField(0) + "--" +
				// item.getField(1));
				// });
				// }
				// })
				;
		env.execute();

	}

	private static class MyFoldFunction implements FoldFunction<String, Tuple3<String, Long, Integer>> {

		public Tuple3<String, Long, Integer> fold(Tuple3<String, Long, Integer> acc, String s) {
			Integer cur = acc.getField(2);
			acc.setField(2, cur + 1);
			return acc;
		}
	}

	private static class MyWindowFunction implements
			WindowFunction<Tuple3<String, Long, Integer>, Tuple3<String, Long, Integer>, String, TimeWindow> {

		public void apply(String key, TimeWindow window, Iterable<Tuple3<String, Long, Integer>> counts,
				Collector<Tuple3<String, Long, Integer>> out) {
			Integer count = counts.iterator().next().getField(2);
			out.collect(new Tuple3<String, Long, Integer>(key, window.getEnd(), count));
		}
	}
}
